package com.reactor.冷热序列;

import java.time.Duration;
import java.time.temporal.ChronoUnit;

import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;

public class Demo2 {

	private static final Logger log = LoggerFactory.getLogger(Demo2.class);

	// 我们可以通过应用响应式转换将冷发布者转换为热发布者，例如，一旦所有订阅者都准备好生成数据，我们可能希望在几个订阅者之间共享冷处理器的结果。
	// 同时，我们又不希望为每个订阅者重新生成数据。reactor为此提供了ConnectableFlux，使用它，不仅可以生成数据以满足最紧迫的需求，还会缓存数据，以便所有其他订阅者可以按照自己的速度处理数据
	// 队列和超时可以通过publish方法和replay方法进行配置。
	@Test
	@Ignore
	public void test1() throws InterruptedException {
		Flux<Long> source = Flux.interval(Duration.of(1, ChronoUnit.SECONDS))
					.take(5)
					.doOnSubscribe(s -> log.info("new subscription for the cold publisher")); // 这里调用了一次，说明只生成了一次数据项，但是每个订阅者都收到了数据集
		ConnectableFlux<Long> conn = source.publish();
		conn.subscribe(e -> log.info("[subscriber 1] onNext: {}", e));
		conn.subscribe(e -> log.info("[subscriber 2] onNext: {}", e));

		log.info("all subscribers are ready, connecting");
		conn.connect();
		Thread.sleep(3000);
		conn.subscribe(e -> log.info("[subscriber 3] onNext: {}", e));

		Thread.sleep(10000);
	}

	// 使用ConnectableFlux可以轻松实现不同的数据缓存策略。但是reactor已经以cache操作符的形式提供了用于事件缓存的api。在内部，cache操作符使用ConnectableFlux，因此
	// 它的主要附加值是他所提供的一个流式而直接的api，我们可以调整缓存所能容纳的数据量以及每个缓存项的到期时间。
	@Test
	@Ignore
	public void test2() throws InterruptedException {
		Flux<Long> source = Flux.interval(Duration.of(100, ChronoUnit.MILLIS))
				.take(5)
				.doOnSubscribe(s -> log.info("new subscription for the cold publisher")); // 创建一个冷发布者
		Flux<Long> cacheSource = source.cache(Duration.ofSeconds(1)); // 缓存冷发布者，缓存时间1s

		cacheSource.subscribe(e -> log.info("[subscriber 1] onNext: {}", e)); // 连接第一个订阅者
		cacheSource.subscribe(e -> log.info("[subscriber 2] onNext: {}", e)); // 连接第二个订阅者

		Thread.sleep(1200); // 等待一段时间以使缓存的数据过期
		cacheSource.subscribe(e -> log.info("[subscriber 3] onNext: {}", e)); // 最后连接第三个订阅者

		Thread.sleep(10000);
	}
	// 根据日志得出结论：前两个订阅者共享第一个订阅的同一份缓存数据。然后在一定延迟之后，由于第三个订阅者无法获取缓存数据，因此一个针对冷发布者的新订阅被触发了。
	// 第三个订阅者接收到了全部数据

	@Test
	//@Ignore
	public void test3() throws InterruptedException {
		Flux<Long> source = Flux.interval(Duration.of(100, ChronoUnit.MILLIS))
				.take(5)
				.doOnSubscribe(s -> log.info("new subscription for the cold publisher"));
		Flux<Long> cacheSource = source.share();

		cacheSource.subscribe(e -> log.info("[subscriber 1] onNext: {}", e)); // 连接第一个订阅者
		Thread.sleep(300);
		cacheSource.subscribe(e -> log.info("[subscriber 2] onNext: {}", e)); // 连接第二个订阅者

		Thread.sleep(10000);
		// 从日志中看到，第一个订阅者从头开始，第二个订阅者错过了在其出现之前所产生的事件
	}

}
